package com.demo.rocketmq.message;

import com.demo.rocketmq.domain.OrderPaidEvent;
import com.demo.rocketmq.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.MimeTypeUtils;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;

/**
 *
 * @author
 * @since
 */
@Component
@Slf4j
public class MessageProducer {
  @Resource
  private RocketMQTemplate rocketMQTemplate;
  @Value("${demo.rocketmq.topic}")
  private String springTopic;
  @Value("${demo.rocketmq.topic.user}")
  private String userTopic;
  @Value("${demo.rocketmq.tag}")
  private String springTopicTag;
  @Value("${demo.rocketmq.orderTopic}")
  private String orderTopic;

  public void sendTestMessage(){
    // Send string
    SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
    log.info("syncSend1 to topic {} sendResult={}", springTopic, sendResult);
    User user = new User();
    user.setUserAge(18);
    user.setUserName("Kitty");
    sendResult = rocketMQTemplate.syncSend(userTopic, user);
    log.info("syncSend1 to topic {} sendResult={} ", userTopic, sendResult);
  }

  public void sendWithMessageBuilder(){

    User user = new User();
    user.setUserAge(19);
    user.setUserName("Kate");

    SendResult sendResult = rocketMQTemplate.syncSend(userTopic, MessageBuilder.withPayload(
        user).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build());
    log.info("syncSend1 to topic {} sendResult={}", userTopic, sendResult);
  }

  public void sendBatchMessage(){
    log.info("批量发送消息开始 ");
    List<Message> msgs = new ArrayList<Message>();
    for (int i = 0; i < 10; i++) {
      msgs.add(MessageBuilder.withPayload("Hello RocketMQ Batch Msg#" + i).
          setHeader(RocketMQHeaders.KEYS, "KEY_" + i).build());
    }

    SendResult sr = rocketMQTemplate.syncSend(springTopic +  ":" + springTopicTag, msgs, 60000);

    log.info("批量发送消息结束，结果:{} ", sr);
  }

  public void sendUserDefineObject(){
    log.info("开始发送用户自定义对象消息");
    rocketMQTemplate.asyncSend(orderTopic, new OrderPaidEvent("T_8888", new BigDecimal("88.88")), new SendCallback() {
      @Override
      public void onSuccess(SendResult sendResult) {
        log.info("send succ. {}", sendResult);
      }

      @Override
      public void onException(Throwable throwable) {
        log.info("send fail. {}", throwable.getMessage());
      }
    });
  }

  public void sendBatchOrderMessage(){
    log.info("----------begin to send batch order message");

    for (int q = 0; q < 4; q++ ) {
      Integer qi = q;
      new Thread(() -> {
        List<Message> messageList = new ArrayList<>(32);
        for (int i = 10; i > 0; i--) {
          int msgIndex = qi * 10 + i;
          String msg = String.format("Hello Rcoketmq batch message#%d to queue:%d", msgIndex, qi);
          messageList.add(MessageBuilder
              .withPayload(msg)
              .setHeader(RocketMQHeaders.KEYS, "KEY_" + msgIndex)
              .build());
        }
        SendResult sr = rocketMQTemplate.syncSendOrderly(springTopic +  ":" + springTopicTag, messageList, qi + "", 60000);
//        SendResult sr = rocketMQTemplate.syncSend(springTopic +  ":" + springTopicTag, messageList,  60000);
        log.info("------batch message orderly send to queue : {} send result :{}", sr.getMessageQueue().getQueueId(), sr);
      }).start();
    }
    log.info("--------All batch order send complete.");
  }

  public void sendTransactionMessage(){
    String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
    for (int i = 0; i < 10; i++) {
      try {

        Message msg = MessageBuilder.withPayload("rocketMQTemplate transactional message " + i).
            setHeader(RocketMQHeaders.TRANSACTION_ID, "KEY_" + i).build();
        SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(
            springTopic + ":" + tags[i % tags.length], msg, null);
        log.info("------rocketMQTemplate send Transactional msg body = {} , sendResult={} {}",
            msg.getPayload(), sendResult.getSendStatus());

        Thread.sleep(10);
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }
}
